package cn.com.pc.content.kafka.producer;

import cn.com.pc.content.domain.*;
import cn.com.pc.content.domain.audit.AuditStatus;
import cn.com.pc.content.domain.content.Author;
import cn.com.pc.content.domain.content.CoverImage;
import cn.com.pc.content.dto.ArticleDTO;
import cn.com.pc.content.util.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

@Component
@Slf4j
public class ContentProducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    private String topic = "pcp-content";
    private String errorLogTopic = "pcp-content-error-log";
    private boolean testKafka = true;

//    @Scheduled(initialDelay = 2000L, fixedDelay = 30 * 24 * 60 * 60 * 1000L)
//    public void testKafka() {
//        log.info("测试kafka发送");
//        if (testKafka) {
//            send("article","referId112",
//                    "","create");
//            testKafka = false;
//        }
//    }

    /**
     * 测试，模拟业务方发生物料内容消息
     * kafka msg key: 内容类型（article/video/ad/product/other）:业务方内容id:物料内容id（新物料没有id则为空）
     *
     * @param contentType
     * @param referId
     * @param contentId
     * @param action
     */
    public void send(String contentType, String referId, String contentId, String action) {

        String key = contentType + ":" + referId + ":" + contentId;

        ArticleDTO articleDTO = new ArticleDTO();
        CoverImage coverImage = new CoverImage();
        coverImage.setDesc("desc");
        coverImage.setUrl("url");
        Author author = new Author();
        author.setPassportId(111L);
        author.setFullName("fullname");
        articleDTO.setId(contentId);
        articleDTO.setSite(Site.PCOnline);
        articleDTO.setBiz(Biz.PGC);
        articleDTO.setApp(App.PGC);
        articleDTO.setReferId(referId);
        articleDTO.setTitle("测试 标题");
        articleDTO.setSummary("测试 简介");
        articleDTO.setKeywords(new HashSet<String>() {{
            add("关键词1");
            add("关键词2");
        }});
        articleDTO.setCoverImages(new ArrayList<CoverImage>() {{
            add(coverImage);
        }});
//        articleDTO.setTagIds(new HashSet<Long>() {{
//            add(103152864L);
//            add(3003L);
//        }});
        articleDTO.setAuthor(author);
        articleDTO.setAuditStatus(AuditStatus.Pending);
        articleDTO.setWeight(10f);
//        articleDTO.setMetaData(new HashMap<String, Object>() {{
////            put("int", new HashSet<Integer>() {{
////                add(1);
////                add(2);
////            }});
//            put("boolean", true);
//        }});
        articleDTO.setStatData(new HashMap<String, Integer>() {{
            put("pvTotal", 1);
            put("cmtCount", 2);
        }});
        articleDTO.setCreatedAt(LocalDateTime.now());
        articleDTO.setCreatedBy("");
        articleDTO.setUpdatedAt(LocalDateTime.now());
        articleDTO.setUpdatedBy("");
        articleDTO.setText("测试内容");

//        CustomizedContentDTO customizedContentDTO = new CustomizedContentDTO();
//        CoverImage coverImage = new CoverImage();
//        coverImage.setDesc("desc");
//        coverImage.setUrl("url");
//        Author author = new Author();
//        author.setPassportId(111L);
//        author.setFullName("fullname");
//        customizedContentDTO.setId(contentId);
//        customizedContentDTO.setSite(Site.PCBaby);
//        customizedContentDTO.setBiz(Biz.CMS);
//        customizedContentDTO.setApp(App.CMS);
//        customizedContentDTO.setReferId(referId);
//        customizedContentDTO.setTitle("测试 标题123");
//        customizedContentDTO.setSummary("测试 简介");
//        customizedContentDTO.setKeywords(new HashSet<String>() {{
//            add("关键词1");
//            add("关键词2");
//        }});
//        customizedContentDTO.setCoverImages(new ArrayList<CoverImage>() {{
//            add(coverImage);
//        }});
//        customizedContentDTO.setTagIds(new HashSet<Long>() {{
//            add(103152864L);
//        }});
//        customizedContentDTO.setAuthor(author);
//        customizedContentDTO.setAuditStatus(AuditStatus.Pending);
//        customizedContentDTO.setWeight(10);
//        customizedContentDTO.setMetaData(new HashMap<String, Object>() {{
////            put("int", new HashSet<Integer>() {{
////                add(1);
////                add(2);
////            }});
//            put("answerContent", "<p>test<align></p>");
//        }});
//        customizedContentDTO.setStatData(new HashMap<String, Integer>() {{
//            put("pvTotal", 1);
//            put("cmtCount", 2);
//        }});
//        customizedContentDTO.setCreatedAt(LocalDateTime.now());
//        customizedContentDTO.setCreatedBy("");
//        customizedContentDTO.setUpdatedAt(LocalDateTime.now());
//        customizedContentDTO.setUpdatedBy("");
//        customizedContentDTO.setContentType(ContentType.Question);

//        VideoDTO videoDTO = new VideoDTO();
//        CoverImage coverImage = new CoverImage();
//        coverImage.setDesc("desc");
//        coverImage.setUrl("url");
//        Author author = new Author();
//        author.setPassportId(111L);
//        author.setFullName("fullname");
//        videoDTO.setId(contentId);
//        videoDTO.setSite(Site.PCOnline);
//        videoDTO.setBiz(Biz.PGC);
//        videoDTO.setApp(App.PGC);
//        videoDTO.setReferId("referId112");
//        videoDTO.setTitle("测试 标题");
//        videoDTO.setSummary("测试 简介");
//        videoDTO.setKeywords(new HashSet<String>() {{
//            add("关键词1");
//            add("关键词2");
//        }});
//        videoDTO.setCoverImages(new ArrayList<CoverImage>() {{
//            add(coverImage);
//        }});
//        videoDTO.setTagIds(new HashSet<Long>() {{
//            add(103152864L);
//        }});
//        videoDTO.setAuthor(author);
//        videoDTO.setAuditStatus(AuditStatus.Pending);
//        videoDTO.setWeight(10);
////        articleDTO.setMetaData(new HashMap<String, Object>() {{
//////            put("int", new HashSet<Integer>() {{
//////                add(1);
//////                add(2);
//////            }});
//////            put("boolean", true);
////        }});
//        videoDTO.setStatData(new HashMap<String, Integer>() {{
//            put("pvTotal", 1);
//            put("cmtCount", 2);
//        }});
//        videoDTO.setCreatedAt(LocalDateTime.now());
//        videoDTO.setCreatedBy("");
//        videoDTO.setUpdatedAt(LocalDateTime.now());
//        videoDTO.setUpdatedBy("");
//        videoDTO.setDuration(100);
//        videoDTO.setSize(1024);

        Map<String, Object> contentMap = new HashMap<>();
        contentMap.put("action", action);
        contentMap.put("content", articleDTO);

        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, key, JsonUtils.obj2String(contentMap));
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            //推送成功
            public void onSuccess(SendResult<String, Object> result) {
                log.debug("发送消息成功:topic({})；msg({})", topic, result.toString());
            }

            @Override
            //推送失败
            public void onFailure(Throwable ex) {
                log.error("发送消息失败:topic({})；errorMsg({})", topic, ex.getMessage());
            }
        });
    }

    /**
     * 发送验证不通过的物料消息，到errorTopic
     *
     * @param msg kafka消息
     */
    @Async
    public void sendToErrorTopic(Object key, String msg) {

        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(errorLogTopic, key, msg);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            //推送成功
            public void onSuccess(SendResult<String, Object> result) {
                log.debug("发送验证失败物料消息成功:topic({})；key({})；msg({})", errorLogTopic, key, result.toString());
            }

            @Override
            //推送失败
            public void onFailure(Throwable ex) {
                log.error("发送验证失败物料消息失败:topic({})；key({})；msg({})", errorLogTopic, key, ex.getMessage());
            }
        });
    }
}
